import Fluent
import FluentSQL
import Queues
import Redis
//
//  DailyStatisticsJobs.swift
//  hello
//
//  Created by 赵康 on 2025/2/16.
//
import Vapor

struct DailyStatisticsJobs: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    _ = app.logger

    try await app.statisticsScheduler.updatePreviousDayStatistics()
  }
}

struct HourlyStatisticsJobs: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    _ = app.logger

    try await app.statisticsScheduler.updateTodayStatistics()
  }
}

struct MonthlyStatisticsJobs: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    _ = app.logger

    try await app.statisticsScheduler.updateMonthlyStatistics()
  }
}
struct RankingCacheJobs: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    try await context.application.statisticsScheduler.updateRankingCache()
  }
}
extension Application {
  var statistics: StatisticsService {
    .init(application: self, cache: StatisticsCache(cache: Cache(logger: logger), redis: redis))
  }
}
extension Application {
  var statisticsScheduler: StatisticsScheduler {
    .init(
      application: self,
      cache: StatisticsCache(cache: Cache(logger: logger), redis: redis)
    )
  }
}
struct DataConsistencyJob: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    _ = app.logger

    // 使用预定义的TimeRange获取过去一周的时间范围
    let timeRange = TimeRange.week
    let (startDate, endDate) = timeRange.getDateRange()

    let users = try await User.query(on: app.db).all()

    // 获取所有标签
    let tags = try await Tag.query(on: app.db).all()

    for user in users {
      guard let userID = user.id else { continue }

      // 1. 清除并重建无标签的统计数据
      try await app.statistics.cache.clearRangeStatistics(
        userID: userID,
        startDate: startDate,
        endDate: endDate,
        tagID: nil
      )

      // 2. 重新计算每天的统计数据
      var currentDate = startDate
      while currentDate < endDate {
        let dayStart = Calendar.mondayFirst.startOfDay(for: currentDate)
        let nextDay = Calendar.mondayFirst.date(byAdding: .day, value: 1, to: dayStart)!

        // 计算无标签统计
        let duration = try await calculateDailyDuration(
          app: app,
          userID: userID,
          startDate: dayStart,
          endDate: nextDay,
          tagID: nil
        )

        // 更新缓存
        try await app.statistics.cache.setDailyStatistics(
          userID: userID,
          date: dayStart,
          tagID: nil,
          duration: duration
        )

        // 3. 处理每个标签的统计
        for tag in tags {
          // 清除标签相关的缓存
          try await app.statistics.cache.clearRangeStatistics(
            userID: userID,
            startDate: dayStart,
            endDate: nextDay,
            tagID: tag.id
          )

          // 重新计算标签统计
          let tagDuration = try await calculateDailyDuration(
            app: app,
            userID: userID,
            startDate: dayStart,
            endDate: nextDay,
            tagID: tag.id
          )

          // 更新标签统计缓存
          try await app.statistics.cache.setDailyStatistics(
            userID: userID,
            date: dayStart,
            tagID: tag.id,
            duration: tagDuration
          )
        }

        currentDate = nextDay
      }
    }

  }

  // 辅助方法：计算每日时长
  private func calculateDailyDuration(
    app: Application,
    userID: UUID,
    startDate: Date,
    endDate: Date,
    tagID: UUID?
  ) async throws -> Double {
    guard let sql = app.db as? SQLDatabase else {
      throw Abort(.internalServerError)
    }

    var query = SQLQueryString(
      """
      		SELECT COALESCE(SUM(h.listenedDuration), 0) as totalDuration
      		FROM \(unsafeRaw: History.schema) h
      """)

    if tagID != nil {
      query += SQLQueryString(
        """
        		JOIN \(unsafeRaw: Audio.schema) a ON h.audio = a.id
        		JOIN \(unsafeRaw: Album.schema) al ON a.album = al.id
        		JOIN \(unsafeRaw: AlbumTag.schema) at ON al.id = at.album
        """)
    }

    query += SQLQueryString(
      """
      		WHERE h.user = \(bind: userID)
      		AND h.lastListenedAt >= \(bind: startDate)
      		AND h.lastListenedAt < \(bind: endDate)
      """)

    if let tagID = tagID {
      query += SQLQueryString(" AND at.tag = \(bind: tagID)")
    }

    let row = try await sql.raw(query).first()
    return try row?.decode(column: "totalDuration", as: Double.self) ?? 0.0
  }
}

struct RecentAudiosStatisticsJob: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    let logger = app.logger

    logger.info("开始更新最近7天音频统计缓存")

    // 计算最近7天的时间范围
    let calendar = Calendar.mondayFirst
    let now = Date()
    let startDate = calendar.date(byAdding: .day, value: -6, to: calendar.startOfDay(for: now))!
    let endDate = calendar.date(byAdding: .day, value: 1, to: calendar.startOfDay(for: now))!

    // 获取所有用户
    let users = try await User.query(on: app.db).all()

    guard let sql = app.db as? SQLDatabase else {
      throw Abort(.internalServerError)
    }

    // 计算所有用户的音频数量分布
    let allUsersQuery = SQLQueryString(
      """
      	SELECT h.user, COUNT(DISTINCT h.audio) as audioCount
      	FROM \(unsafeRaw: History.schema) h
      	WHERE h.lastListenedAt >= \(bind: startDate)
      	AND h.lastListenedAt < \(bind: endDate)
      	AND h.listenedDuration > 0
      	GROUP BY h.user
      """)

    let allUsersRows = try await sql.raw(allUsersQuery).all()
    let allUsersCounts = try allUsersRows.map { row in
      try row.decode(column: "audioCount", as: Int.self)
    }

    // 更新全局分布缓存
    try await app.statistics.cache.setAllUsersRecentAudiosDistribution(distribution: allUsersCounts)

    // 为每个用户更新个人缓存
    for user in users {
      guard let userID = user.id else { continue }

      // 查询每个用户的音频数量
      let userQuery = SQLQueryString(
        """
        	SELECT COUNT(DISTINCT h.audio) as audioCount
        	FROM \(unsafeRaw: History.schema) h
        	WHERE h.user = \(bind: userID)
        	AND h.lastListenedAt >= \(bind: startDate)
        	AND h.lastListenedAt < \(bind: endDate)
        	AND h.listenedDuration > 0
        """)

      let userRow = try await sql.raw(userQuery).first()
      let userAudioCount = try userRow?.decode(column: "audioCount", as: Int.self) ?? 0

      // 更新用户缓存
      try await app.statistics.cache.setRecentAudiosCount(userID: userID, count: userAudioCount)
    }

    logger.info("最近7天音频统计缓存更新完成")
  }
}

struct TagsDistributionJob: AsyncScheduledJob {
  func run(context: QueueContext) async throws {
    let app = context.application
    let logger = app.logger

    logger.info("开始更新标签分布缓存")

    // 获取所有用户
    let users = try await User.query(on: app.db).all()

    // 为每个活跃用户预热标签分布缓存
    for user in users {
      guard let userID = user.id else { continue }

      // 检查该用户在最近7天是否有收听记录
      let calendar = Calendar.mondayFirst
      let now = Date()
      let startDate = calendar.date(byAdding: .day, value: -6, to: calendar.startOfDay(for: now))!
      let endDate = calendar.date(byAdding: .day, value: 1, to: calendar.startOfDay(for: now))!

      let hasActivity =
        try await History.query(on: app.db)
        .filter(\.$user.$id == userID)
        .filter(\.$lastListenedAt >= startDate)
        .filter(\.$lastListenedAt < endDate)
        .first() != nil

      // 只为有活动的用户更新缓存
      if hasActivity {
        // 强制更新缓存（先删除旧缓存）
        try await app.statistics.cache.delete(
          forKey: CacheConfig.KeyBuilder.tagsDistribution(userID: userID),
          on: app.redis
        )

        // 重新计算并缓存标签分布
        _ = try await app.statistics.getTagsDistribution(userID: userID)
        logger.debug("已更新用户 \(userID) 的标签分布缓存")
      }
    }

    logger.info("标签分布缓存更新完成")
  }
}
